package org.example.listener;

import com.alibaba.fastjson2.JSONObject;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.example.mapper.AccountInfoMapper;
import org.example.model.message.AccountChangeMessage;
import org.example.service.AccountInfoService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

/**
 * @author shenguangyang
 */
@Component
@RocketMQTransactionListener
public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {
    private static final Logger log = LoggerFactory.getLogger(ProducerTxmsgListener.class);
    @Resource
    private AccountInfoService accountInfoService;

    @Resource
    private AccountInfoMapper accountInfoMapper;

    /**
     * 消息发送成功回调此方法，此方法执行本地事务
     */
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        // 解析消息内容
        try {
            String jsonString = new String((byte[]) message.getPayload());
            AccountChangeMessage accountChangeMessage = JSONObject.parseObject(jsonString, AccountChangeMessage.class);
            // 扣除金额
            accountInfoService.doUpdateAccountBalance(accountChangeMessage);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("executeLocalTransaction 事务执行失败", e);
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 此方法检查事务执行状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        RocketMQLocalTransactionState state;
        AccountChangeMessage accountChangeMessage = JSONObject.parseObject(new String((byte[]) message.getPayload()), AccountChangeMessage.class);
        // 事务id
        String txNo = accountChangeMessage.getMsgId();
        int isExistTx = accountInfoMapper.isExistTx(txNo);
        log.info("回查事务, 事务号: {} 结果: {}", accountChangeMessage.getMsgId(), isExistTx);
        if (isExistTx > 0) {
            state = RocketMQLocalTransactionState.COMMIT;
        } else {
            // 未知, 还会重新回查
            state = RocketMQLocalTransactionState.UNKNOWN;
        }
        return state;
    }
}
